package com.rocketmq;



import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.common.message.MessageExt;
import org.junit.Test;

import java.nio.charset.StandardCharsets;
import java.util.List;

public class MyConsumer {
    @Test
    public void consume() throws Exception {
    //1 创建一个消费者 消费类型2种 push 连接nameserver
        //创建消费对象
        DefaultMQPushConsumer consumer=new DefaultMQPushConsumer();
        //填充属性 nameserver 消费者分组 监听的主题 
        consumer.setNamesrvAddr("localhost:9876");
        consumer.setConsumerGroup("c-group01");
        //如果消费者启动的时候 主题不存在 消费者会抛出一个error 提示 route路由找不到主题
        consumer.subscribe("topic01","*");
    //2 创建一个消息监听器 随时等待被push的消息执行消费逻辑 涉及到消息的序列化 反序列化问题
        //准备一个监听器
        MessageListenerConcurrently messageListener=
                new MessageListenerConcurrently() {
                    /**每接收到一条消息,就要调用一次consumeMessage方法 处理消费逻辑
                     * @param list 接收到的消息 每次只消费一条消息 list元素只有一个
                     * @param consumeConcurrentlyContext 监听器上下文,只有业务处理消息的话,基本用不到
                     * @return 消费处理结果 只有2种 CONSUME_SUCCESS RECONSUME_LATER
                     */
                    @Override
                    public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> list, ConsumeConcurrentlyContext consumeConcurrentlyContext) {
                        //测试案例的消费逻辑 将消息中的new byte[] 消息体拿出来 做解析 打印 
                        //返回消费成功
                        MessageExt messageExt = list.get(0);
                        String msgId = messageExt.getMsgId();
                        byte[] body = messageExt.getBody();
                        System.out.println("消费者01接到消息id:"+msgId);
                        String message=new String(body, StandardCharsets.UTF_8);
                        System.out.println("消费者01接到消息msg:"+message);
                        if(message!=null&&message.equals("reconsume")){
                            return ConsumeConcurrentlyStatus.RECONSUME_LATER;
                        }
                        return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
                    }
                };
        consumer.setMessageListener(messageListener);
    //3 消费者要想看到消费逻辑,保证当前线程不死
        consumer.start();//开启连接
        while(true);
    }
    @Test
    public void consume01() throws Exception {
        //1 创建一个消费者 消费类型2种 push 连接nameserver
        //创建消费对象
        DefaultMQPushConsumer consumer=new DefaultMQPushConsumer();
        //填充属性 nameserver 消费者分组 监听的主题
        consumer.setNamesrvAddr("localhost:9876");
        consumer.setConsumerGroup("c-group02");
        //如果消费者启动的时候 主题不存在 消费者会抛出一个error 提示 route路由找不到主题
        consumer.subscribe("topic01","*");
        //2 创建一个消息监听器 随时等待被push的消息执行消费逻辑 涉及到消息的序列化 反序列化问题
        //准备一个监听器
        MessageListenerConcurrently messageListener=
                new MessageListenerConcurrently() {
                    /**每接收到一条消息,就要调用一次consumeMessage方法 处理消费逻辑
                     * @param list 接收到的消息 每次只消费一条消息 list元素只有一个
                     * @param consumeConcurrentlyContext 监听器上下文,只有业务处理消息的话,基本用不到
                     * @return 消费处理结果 只有2种 CONSUME_SUCCESS RECONSUME_LATER
                     */
                    @Override
                    public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> list, ConsumeConcurrentlyContext consumeConcurrentlyContext) {
                        //测试案例的消费逻辑 将消息中的new byte[] 消息体拿出来 做解析 打印
                        //返回消费成功
                        MessageExt messageExt = list.get(0);
                        String msgId = messageExt.getMsgId();
                        byte[] body = messageExt.getBody();
                        System.out.println("消费者01接到消息id:"+msgId);
                        System.out.println("消费者01接到消息msg:"+new String(body, StandardCharsets.UTF_8));
                        return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
                    }
                };
        consumer.setMessageListener(messageListener);
        //3 消费者要想看到消费逻辑,保证当前线程不死
        consumer.start();//开启连接
        while(true);
    }
    @Test
    public void consume02() throws Exception {
        //1 创建一个消费者 消费类型2种 push 连接nameserver
        //创建消费对象
        DefaultMQPushConsumer consumer=new DefaultMQPushConsumer();
        //填充属性 nameserver 消费者分组 监听的主题
        consumer.setNamesrvAddr("localhost:9876");
        consumer.setConsumerGroup("c-group03");
        //如果消费者启动的时候 主题不存在 消费者会抛出一个error 提示 route路由找不到主题
        consumer.subscribe("topic01","*");
        //2 创建一个消息监听器 随时等待被push的消息执行消费逻辑 涉及到消息的序列化 反序列化问题
        //准备一个监听器
        MessageListenerConcurrently messageListener=
                new MessageListenerConcurrently() {
                    /**每接收到一条消息,就要调用一次consumeMessage方法 处理消费逻辑
                     * @param list 接收到的消息 每次只消费一条消息 list元素只有一个
                     * @param consumeConcurrentlyContext 监听器上下文,只有业务处理消息的话,基本用不到
                     * @return 消费处理结果 只有2种 CONSUME_SUCCESS RECONSUME_LATER
                     */
                    @Override
                    public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> list, ConsumeConcurrentlyContext consumeConcurrentlyContext) {
                        //测试案例的消费逻辑 将消息中的new byte[] 消息体拿出来 做解析 打印
                        //返回消费成功
                        MessageExt messageExt = list.get(0);
                        String msgId = messageExt.getMsgId();
                        byte[] body = messageExt.getBody();
                        System.out.println("消费者01接到消息id:"+msgId);
                        System.out.println("消费者01接到消息msg:"+new String(body, StandardCharsets.UTF_8));
                        return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
                    }
                };
        consumer.setMessageListener(messageListener);
        //3 消费者要想看到消费逻辑,保证当前线程不死
        consumer.start();//开启连接
        while(true);
    }
    @Test
    public void consume03() throws Exception {
        //1 创建一个消费者 消费类型2种 push 连接nameserver
        //创建消费对象
        DefaultMQPushConsumer consumer=new DefaultMQPushConsumer();
        //填充属性 nameserver 消费者分组 监听的主题
        consumer.setNamesrvAddr("localhost:9876");
        consumer.setConsumerGroup("c-group04");
        //如果消费者启动的时候 主题不存在 消费者会抛出一个error 提示 route路由找不到主题
        consumer.subscribe("topic01","上海");
        //2 创建一个消息监听器 随时等待被push的消息执行消费逻辑 涉及到消息的序列化 反序列化问题
        //准备一个监听器
        MessageListenerConcurrently messageListener=
                new MessageListenerConcurrently() {
                    /**每接收到一条消息,就要调用一次consumeMessage方法 处理消费逻辑
                     * @param list 接收到的消息 每次只消费一条消息 list元素只有一个
                     * @param consumeConcurrentlyContext 监听器上下文,只有业务处理消息的话,基本用不到
                     * @return 消费处理结果 只有2种 CONSUME_SUCCESS RECONSUME_LATER
                     */
                    @Override
                    public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> list, ConsumeConcurrentlyContext consumeConcurrentlyContext) {
                        //测试案例的消费逻辑 将消息中的new byte[] 消息体拿出来 做解析 打印
                        //返回消费成功
                        MessageExt messageExt = list.get(0);
                        String msgId = messageExt.getMsgId();
                        byte[] body = messageExt.getBody();
                        /*TeduUser user = convert(body);*/
                        System.out.println("消费者01接到消息id:"+msgId);
                        System.out.println("消费者01接到消息msg:"+new String(body, StandardCharsets.UTF_8));
                        return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
                    }
                };
        consumer.setMessageListener(messageListener);
        //3 消费者要想看到消费逻辑,保证当前线程不死
        consumer.start();//开启连接
        while(true);
    }
}
